package scheduler

import (
	"context"
	"fmt"
	"sync"

	logger "gitee.com/youkelike/zlog"
	"gitee.com/youkelike/zmigrate/migrator/events"
	"gitee.com/youkelike/zmigrate/migrator/validator"
	"gitee.com/youkelike/zmigrate/pkg/ginx"
	"gitee.com/youkelike/zmigrate/pkg/gormx/connpool"
	"github.com/gin-gonic/gin"
	"golang.org/x/sync/errgroup"
)

// Scheduler 用来统一管理一张表的整个迁移过程，
// 每个表对应一个 Scheduler，
// 只能控制所在实例上的迁移过程，
// 在多实例运行的场景下，每个实例都有自己的 Scheduler，它们都是独立的，
// 如果要启动全量校验、增量校验，只需启动任意一个节点上的 http 接口，多个节点同时运行校验没有意义，
// 但是如果要改变 pattern 值控制双写，每个节点上的对应 http 接口都需要请求一次,
type Scheduler struct {
	lock       sync.Mutex
	pool       *connpool.DoubleWritePool
	l          logger.Logger
	pattern    string
	cancelFull []func()

	// canal 的消费者
	consumers []events.CanalConsumer
	// 全量校验器
	fullValidators []validator.Validator
}

// 每张需要迁移的表对应一个这样的实例，
func NewScheduler(l logger.Logger, pool *connpool.DoubleWritePool) *Scheduler {
	return &Scheduler{
		l:          l,
		pattern:    connpool.PatternSrcOnly,
		cancelFull: make([]func(), 5),
		pool:       pool,
	}
}

func (s *Scheduler) SetCanalConsumers(consumers []events.CanalConsumer) {
	s.consumers = consumers
}
func (s *Scheduler) SetFullValidators(validators []validator.Validator) {
	s.fullValidators = validators
}

// 这一个也不是必须的，就是你可以考虑利用配置中心，监听配置中心的变化
// 把全量校验，增量校验做成分布式任务，利用分布式任务调度平台来调度
func (s *Scheduler) RegisterRoutes(server *gin.RouterGroup) {
	// 将这个暴露为 HTTP 接口
	// 你可以配上对应的 UI
	server.POST("/src_only", ginx.Wrap(s.l, s.SrcOnly))
	server.POST("/src_first", ginx.Wrap(s.l, s.SrcFirst))
	server.POST("/dst_first", ginx.Wrap(s.l, s.DstFirst))
	server.POST("/dst_only", ginx.Wrap(s.l, s.DstOnly))
	server.POST("/full/start", ginx.Wrap(s.l, s.StartFullValidation))
	server.POST("/full/stop", ginx.Wrap(s.l, s.StopFullValidation))
	server.POST("/incr/stop", ginx.Wrap(s.l, s.StopIncrementValidation))
	server.POST("/incr/start", ginx.Wrap(s.l, s.StartIncrementValidation))
}

// ---- 下面是四个阶段 ---- //

// SrcOnly 只读写源表
func (s *Scheduler) SrcOnly(c *gin.Context) (ginx.Result, error) {
	s.lock.Lock()
	defer s.lock.Unlock()
	s.pattern = connpool.PatternSrcOnly
	s.pool.UpdatePattern(connpool.PatternSrcOnly)
	return ginx.Result{
		Msg: "OK",
	}, nil
}

// 先开增量校验再开双写
func (s *Scheduler) SrcFirst(c *gin.Context) (ginx.Result, error) {
	err := s.startIncrementValidation()
	if err != nil {
		return ginx.Result{
			Msg: fmt.Sprintf("启动增量校验失败: %v", err),
		}, nil
	}

	s.lock.Lock()
	s.pattern = connpool.PatternSrcFirst
	s.pool.UpdatePattern(connpool.PatternSrcFirst)
	defer s.lock.Unlock()

	return ginx.Result{
		Msg: "OK",
	}, nil
}

// 切换双写方向的同时切换增量校验的方向
func (s *Scheduler) DstFirst(c *gin.Context) (ginx.Result, error) {
	// 先确保已停掉全量校验
	s.stopFullValidation()

	s.lock.Lock()
	s.pattern = connpool.PatternDstFirst
	s.pool.UpdatePattern(connpool.PatternDstFirst)
	s.lock.Unlock()

	for _, v := range s.consumers {
		v.DstFirst()
	}

	return ginx.Result{
		Msg: "OK",
	}, nil
}

// 停止双写后停止增量校验
func (s *Scheduler) DstOnly(c *gin.Context) (ginx.Result, error) {
	s.lock.Lock()
	s.pattern = connpool.PatternDstOnly
	s.pool.UpdatePattern(connpool.PatternDstOnly)
	s.lock.Unlock()

	_ = s.stopIncrementValidation()

	return ginx.Result{
		Msg: "OK",
	}, nil
}

func (s *Scheduler) StopFullValidation(c *gin.Context) (ginx.Result, error) {
	s.stopFullValidation()

	return ginx.Result{
		Msg: "OK",
	}, nil
}
func (s *Scheduler) stopFullValidation() {
	s.lock.Lock()
	defer s.lock.Unlock()
	for _, v := range s.cancelFull {
		if v != nil {
			v()
		}
	}
}

// 全量校验要手动启动，
// 仅在 src_only/src_first 双写方向下才允许全量校验
func (s *Scheduler) StartFullValidation(c *gin.Context) (ginx.Result, error) {
	// 取消上一次的
	s.stopFullValidation()

	s.lock.Lock()
	defer s.lock.Unlock()
	if s.pattern != connpool.PatternSrcOnly && s.pattern != connpool.PatternSrcFirst {
		return ginx.Result{
			Msg: "仅在 src_only/src_first 双写方向下才允许全量校验",
		}, nil
	}

	for i, v := range s.fullValidators {
		ctx, cancelFull := context.WithCancel(context.Background())
		s.cancelFull[i] = cancelFull
		go func(ctx context.Context, v validator.Validator) {
			err := v.Validate(ctx)
			if err != nil {
				s.l.Warnw("退出全量校验", "err", err)
			}
		}(ctx, v)
	}

	return ginx.Result{
		Msg: "OK",
	}, nil
}

// 增量校验用 canal，在这里启动 canal 消费者
// 可以在 src_only、src_first、dst_first 双写方向下启动增量校验，
// 切换到 dst_only 时会先关闭增量校验，
func (s *Scheduler) StartIncrementValidation(c *gin.Context) (ginx.Result, error) {
	err := s.startIncrementValidation()
	if err != nil {
		return ginx.Result{
			Msg: fmt.Sprintf("启动增量校验失败: %v", err),
		}, nil
	}

	return ginx.Result{
		Msg: "启动增量校验成功",
	}, nil
}
func (s *Scheduler) startIncrementValidation() error {
	var eg errgroup.Group
	for _, v := range s.consumers {
		eg.Go(func() error {
			return v.Start()
		})
	}
	return eg.Wait()
}

// 手动退出增量校验
func (s *Scheduler) StopIncrementValidation(c *gin.Context) (ginx.Result, error) {
	err := s.stopIncrementValidation()
	if err != nil {
		return ginx.Result{
			Msg: fmt.Sprintf("退出增量校验失败: %v", err),
		}, nil
	}
	return ginx.Result{
		Msg: "OK",
	}, nil
}
func (s *Scheduler) stopIncrementValidation() error {
	var eg errgroup.Group
	for _, v := range s.consumers {
		eg.Go(func() error {
			return v.Stop()
		})
	}
	return eg.Wait()
}
